package com.huan.kafka.api;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

/**
 * 测试kafka发送者
 * 1、KafkaProducer线程安全的，可以在多线程中使用。
 * 2、消息发送的key和value的序列化
 * 3、自定义分区的使用
 * 4、自定义拦截器的使用
 * 5、消息发送完成后的回调使用
 *
 * @author huan.fu 2021/1/4 - 上午10:11
 */
public class KafkaProducerDemo {

    public static void main(String[] args) {
        Properties properties = new Properties();
        // kafka服务器的地址，端口，可以不用全写，写几个既可
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092,127.0.0.1:9093,127.0.0.1:9094");
        // kafka 消息key的序列化方式
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // kafka 消息value的序列化方式
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // 0: 消息只要发送出去了，就认为是发送成功
        // 1: 消息被 leader 分区保存后，就认为发送成功
        // all: 消息被 leader + isr 都保存成功后，才认为发送成功
        properties.put(ProducerConfig.ACKS_CONFIG, "all");
        // 设置消息重试的次数
        properties.put(ProducerConfig.RETRIES_CONFIG, 0);
        // 消息发送失败后，重试的间隔，即等待多少毫秒后在重试
        properties.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 300);
        // kafka 的消息默认是批量发送，即等批次消息达到一定大小(单位字节)的时候发送，此参数控制批次消息的大小
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        // batch.size 设置kafka消息批次达到一定的字节时才发送，那万一一直没有达到，此时就由linger.ms控制，即等待多少毫秒之后发送，等于0表示立即发送
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 100);
        // 表示生产者可以用来缓存等待发送到服务器上消息的可用内存大小。
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        // 控制一次请求的大小，防止出现一个非常大的请求，默认是一兆
        properties.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 1048576);
        // 控制客户端等待服务端响应的最大时间，此配置的值应该要大于 replica.lag.time.max(服务端移除isr的一个超时配置) 的值。
        properties.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
        // 配置控制KafkaProducer.send()和KafkaProducer.partitionsFor()会阻塞多长时间。这些方法可能因为缓冲区已满或元数据不可用而被阻塞。
        properties.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 60000);
        // 配置kafka自定义的分区器，我们自己实现的分区器，需要实现 org.apache.kafka.clients.producer.Partitioner 接口
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "com.huan.kafka.api.CustomPartitioner");
        // 配置kafka自定义的拦截器,可以配置多个,多个以英文的逗号分割开
        properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, "com.huan.kafka.api.CustomProducerInterceptor");

        // KafkaProducer 是线程安全的，可以多个线程使用用一个 KafkaProducer
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        for (int i = 0; i < 10; i++) {
            ProducerRecord<String, String> record = new ProducerRecord<>("topic-a", "value - (" + i + 1 + ")");
            kafkaProducer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception != null) {
                        System.err.println("发送数据到kafka中,发生了异常.");
                        exception.printStackTrace();
                        return;
                    }
                    System.out.println("topic: " + metadata.topic() + " offset: " + metadata.offset() + " partition: "
                            + metadata.partition());
                }
            });
        }

        System.out.println("消息发送完成");
        kafkaProducer.close();
    }
}
